package com.tn.rtdata;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Iterator;
import java.util.Map;
import java.util.Queue;
import java.util.UUID;
import java.util.concurrent.ConcurrentLinkedQueue;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * The base class that provides the common behaviour required by fills.
 */
public abstract class Fill<K, V> implements Serializable
{
  private static final Logger LOGGER = LoggerFactory.getLogger(Fill.class);

  private Map<String, DataEventStats> dataEventStatsCache;
  private FillIdentity fillIdentity;
  private Map<FillIdentity, FillInfo<K, V>> fillInfoCache;
  private Queue<DataEvent<V>> queuedDataEvents;
  private Collection<FillListener<V>> fillListeners;

  /**
   * Creates a new <code>Fill</code>.
   *
   * @param fillIdentity        the fill identity.
   * @param fillInfoCache       the cache the fill info will be held in.
   * @param dataEventStatsCache the cache the data event stats will be held in.
   */
  public Fill(
    FillIdentity fillIdentity,
    Map<FillIdentity, FillInfo<K, V>> fillInfoCache,
    Map<String, DataEventStats> dataEventStatsCache
  )
  {
    this.fillIdentity = fillIdentity;
    this.fillInfoCache = fillInfoCache;
    this.dataEventStatsCache = dataEventStatsCache;
    this.queuedDataEvents = new ConcurrentLinkedQueue<DataEvent<V>>();
    this.fillListeners = new ArrayList<FillListener<V>>();
  }

  /**
   * Returns the fill identity.
   */
  public FillIdentity getFillIdentity()
  {
    return fillIdentity;
  }

  /**
   * Returns <code>true</code> if the fill has <code>FillListener</code>s
   */
  public boolean hasListeners()
  {
    return !getListeners().isEmpty();
  }

  /**
   * Returns the started flag.
   */
  public boolean isStarted()
  {
    return fillInfoCache.containsKey(fillIdentity);
  }

  /**
   * Adds the <code>fillListener</code>, loading the fill if need be and calling the
   * <code>MessageProducingFillListener.fillInitialized(Set)</code> method.
   *
   * @throws FillException if an error occurs loading the fill.
   */
  public void addListener(FillListener<V> fillListener) throws FillException
  {
    lock();

    try
    {
      if (isStarted())
      {
        LOGGER.debug("Firing fill initialized for fill: {}", fillIdentity);
        fillListener.fillInitialized(createEventId(DataEvent.Type.INITIAL), values());
      }

      Collection<FillListener<V>> fillListeners = getListeners();
      fillListeners.add(fillListener);
      setFillListeners(fillListeners);
    }
    finally
    {
      unlock();
    }
  }

  /**
   * Removes the <code>fillListener</code>, clearing the fill if there are no more <code>MessageProducingFillListener</code>s for
   * this <code>Fill</code>.
   */
  public void removeListener(FillListener fillListener) throws FillException
  {
    lock();

    try
    {
      Collection<FillListener<V>> fillListeners = getListeners();
      fillListeners.remove(fillListener);
      setFillListeners(fillListeners);

      if (fillListeners.isEmpty())
      {
        LOGGER.debug("Last listener removed, destroying fill: {}", fillIdentity);
        stop();
      }
    }
    finally
    {
      unlock();
    }
  }

  /**
   * Starts the fill.
   *
   * @throws FillException if an error occurs starting the fill.
   */
  public final void start() throws FillException
  {
    lock();

    try
    {
      if (!fillInfoCache.containsKey(fillIdentity))
      {
        beforeStart();

        LOGGER.debug("Initializing fill: {}", fillIdentity);
        Map<K, V> fill = initializeFill();

        fillInfoCache.put(fillIdentity, new FillInfo<K, V>(fillIdentity, new ArrayList<K>(fill.keySet()), fillListeners));
        fillListeners.clear();

        LOGGER.debug("Firing fill initialized for fill: {}", fillIdentity);
        fireInitialized(fill.values());
        cleanQueuedDataEvents(fill.values());

        afterStart();
      }
    }
    finally
    {
      fireQueuedDataEvents();
      unlock();
    }
  }

  /**
   * Stops the fill.
   *
   * @throws FillException if an error occurs starting the fill.
   */
  public final void stop() throws FillException
  {
    lock();

    try
    {
      if (fillInfoCache.containsKey(fillIdentity))
      {
        beforeStop();

        fillInfoCache.remove(fillIdentity);

        afterStop();
      }
    }
    finally
    {
      unlock();
    }
  }

  /**
   * Returns the identity for the <code>value</code>.
   */
  protected abstract K getIdentity(V value);

  /**
   * Returns the <code>MessageProducingFillListener</code>s for this <code>Fill</code>.
   */
  protected Collection<FillListener<V>> getListeners()
  {
    FillInfo<K, V> fillInfo = fillInfoCache.get(fillIdentity);
    return fillInfo != null ? fillInfo.getListeners() : fillListeners;
  }

  /**
   * Returns the locked flag which is <code>true</code> when the <code>lock()</code> method has been called but not the <code>unlock</code>
   * method; otherwise <code>false</code>.
   */
  protected abstract boolean isLocked();

  /**
   * Invoked after this <code>Fill</code> has been started.
   */
  protected void afterStart()
  {
    //Intentionally blank - see JavaDoc.
  }

  /**
   * Invoked before this <code>Fill</code> has been started.
   */
  protected void beforeStart()
  {
    //Intentionally blank - see JavaDoc.
  }

  /**
   * Invoked after this <code>Fill</code> has been stopped.
   */
  protected void afterStop()
  {
    //Intentionally blank - see JavaDoc.
  }

  /**
   * Invoked before this <code>Fill</code> has been stopped.
   */
  protected void beforeStop()
  {
    //Intentionally blank - see JavaDoc.
  }

  /**
   * Invoked when this fill is being destroyed.
   */
  protected void destroy()
  {
    //Intentionally blank - this method is optionally overridden to perform clean-up when the fill is destroyed.
  }

  /**
   * Acquires the lock for this <code>Fill</code> or blocks until the lock is available.
   */
  protected abstract void lock();

  /**
   * Releases the lock for this <code>Fill</code>.
   */
  protected abstract void unlock();

  /**
   * Returns the fillEntries for this <code>Fill</code>.
   */
  protected abstract Collection<V> values();

  /**
   * Fires the <code>FillListener.entryAdded(T)</code> callback method.
   */
  protected void fireEntryAdded(V newEntry)
  {
    if (hasListeners() && queuedDataEvents == null)
    {
      lock();

      try
      {
        FillInfo<K, V> fillInfo = fillInfoCache.get(fillIdentity);
        fillInfo.getContents().add(getIdentity(newEntry));
        fillInfoCache.put(fillIdentity, fillInfo);

        String eventId = createEventId(DataEvent.Type.ADD);

        LOGGER.debug("Firing update event for fill: {}", fillIdentity);
        for (FillListener<V> fillListener : getListeners())
        {
          fillListener.entryAdded(eventId, newEntry);
        }
      }
      finally
      {
        unlock();
      }
    }
    else if (queuedDataEvents != null)
    {
      LOGGER.debug("Queuing event during initialization for fill: {}", fillIdentity);
      queuedDataEvents.add(new DataEvent<V>(null, DataEvent.Type.ADD, newEntry, null));
    }
  }

  /**
   * Fires the <code>FillListener.entryDeleted(T)</code> callback method.
   */
  protected void fireEntryDeleted(V oldEntry)
  {
    if (hasListeners() && queuedDataEvents == null)
    {
      lock();

      try
      {
        FillInfo<K, V> fillInfo = fillInfoCache.get(fillIdentity);
        fillInfo.getContents().remove(getIdentity(oldEntry));
        fillInfoCache.put(fillIdentity, fillInfo);

        String eventId = createEventId(DataEvent.Type.DELETE);

        LOGGER.debug("Firing update event for fill: {}", fillIdentity);
        for (FillListener<V> fillListener : getListeners())
        {
          fillListener.entryDeleted(eventId, oldEntry);
        }
      }
      finally
      {
        unlock();
      }
    }
    else if (queuedDataEvents != null)
    {
      LOGGER.debug("Queuing event during initialization for fill: {}", fillIdentity);
      queuedDataEvents.add(new DataEvent<V>(null, DataEvent.Type.DELETE, null, oldEntry));
    }
  }

  /**
   * Fires the <code>FillListener.fillInitialized(Collection<T>)</code> callback method.
   */
  private void fireInitialized(Collection<V> fillEntries)
  {
    lock();

    try
    {
      String eventId = createEventId(DataEvent.Type.INITIAL);

      LOGGER.debug("Firing initialize event for fill: {}", fillIdentity);
      for (FillListener<V> fillListener : getListeners())
      {
        fillListener.fillInitialized(eventId, fillEntries);
      }
    }
    finally
    {
      unlock();
    }
  }

  /**
   * Fires the <code>FillListener.entryUpdated(T, T)</code> callback method.
   */
  protected void fireEntryUpdated(V newEntry, V oldEntry)
  {
    if (hasListeners() && queuedDataEvents == null)
    {
      lock();

      try
      {
        String eventId = createEventId(DataEvent.Type.UPDATE);

        LOGGER.debug("Firing update event for fill: {}", fillIdentity);
        for (FillListener<V> fillListener : getListeners())
        {
          fillListener.entryUpdated(eventId, newEntry, oldEntry);
        }
      }
      finally
      {
        unlock();
      }
    }
    else if (queuedDataEvents != null)
    {
      LOGGER.debug("Queuing event during initialization for fill: {}", fillIdentity);
      queuedDataEvents.add(new DataEvent<V>(null, DataEvent.Type.UPDATE, newEntry, oldEntry));
    }
  }

  /**
   * Loads the fill data.
   *
   * @return the fill data.
   *
   * @throws FillException if an error occurs during the load.
   */
  protected abstract Map<K, V> initializeFill() throws FillException;

  /**
   * Sets the <code>MessageProducingFillListener</code>s for this <code>Fill</code>.
   */
  private void setFillListeners(Collection<FillListener<V>> fillListeners)
  {
    FillInfo<K, V> fillInfo = fillInfoCache.get(fillIdentity);
    if (fillInfo != null)
    {
      fillInfoCache.put(fillIdentity, new FillInfo<K, V>(fillIdentity, fillInfo.getContents(), fillListeners));
    }
    else
    {
      this.fillListeners = fillListeners;
    }
  }

  /**
   * Creates a new data event id, that is used to track data event processing.
   */
  private String createEventId(DataEvent.Type dataEventType)
  {
    String eventId = UUID.randomUUID().toString();
    dataEventStatsCache.put(eventId, new DataEventStats(eventId, fillIdentity.getName(), dataEventType));

    return eventId;
  }

  /**
   * Removes any queued <code>DataEvent</code>s that are the result of fill initialization.
   */
  private void cleanQueuedDataEvents(Collection<V> values)
  {
    Iterator<DataEvent<V>> dataEventIterator = queuedDataEvents.iterator();
    while (dataEventIterator.hasNext())
    {
      DataEvent<V> dataEvent = dataEventIterator.next();
      if (values.contains(dataEvent.getNewValue()))
      {
        dataEventIterator.remove();
      }
    }
  }

  /**
   * Fires any queued updates that were received while this <code>Fill</code> was initializing.
   */
  private void fireQueuedDataEvents()
  {
    if (queuedDataEvents != null)
    {
      LOGGER.debug("Firing {} queued data events: {}", queuedDataEvents.size(), queuedDataEvents);

      while (!queuedDataEvents.isEmpty())
      {
        DataEvent<V> dataEvent = queuedDataEvents.poll();

        if (dataEvent != null)
        {
          switch (dataEvent.getType())
          {
            case ADD:
              fireEntryAdded(dataEvent.getNewValue());
              return;

            case UPDATE:
              fireEntryUpdated(dataEvent.getNewValue(), dataEvent.getOldValue());
              return;

            case DELETE:
              fireEntryDeleted(dataEvent.getOldValue());
              return;

            default:
              LOGGER.warn("Unrecognized data event type: {}", dataEvent.getType());
          }
        }
      }

      //Make sure there are no more queue data events.
      if (queuedDataEvents.isEmpty())
      {
        queuedDataEvents = null;
        LOGGER.debug("All queued data events fired, processing switched to real-time");
      }
      else
      {
        fireQueuedDataEvents();
      }
    }
  }
}
